package com.mall.order.mq.delay;

import com.mall.order.dal.entitys.Order;
import com.mall.order.dal.entitys.OrderItem;
import com.mall.order.dal.entitys.Stock;
import com.mall.order.dal.persistence.OrderItemMapper;
import com.mall.order.dal.persistence.OrderMapper;
import com.mall.order.dal.persistence.StockMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import tk.mybatis.mapper.entity.Example;

import javax.annotation.PostConstruct;
import java.io.UnsupportedEncodingException;
import java.util.List;

/**
 * @author hanshuai
 * @version 1.0
 * @description @TODO
 * @date 2021/12/15 19:54
 */
@Component
@Slf4j
public class DelayOrderConsumer {
    private DefaultMQPushConsumer consumer;
    @Autowired
    private OrderMapper orderMapper;
    @Autowired
    private OrderItemMapper orderItemMapper;
    @Autowired
    private StockMapper stockMapper;

    @PostConstruct
    public void init() {
        consumer = new DefaultMQPushConsumer("delayOrderConsumer");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        //订阅 消息
        try {
            consumer.subscribe("delayOrder", "*");
            consumer.setMessageListener(new MessageListenerConcurrently() {
                @Override
                @Transactional
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    System.out.println("hahah");
                    //获取orderId
                    MessageExt message = list.get(0);
                    byte[] body = message.getBody();
                    try {
                        String orderId = new String(body, 0, body.length, "utf-8");
                        Example example = new Example(Order.class);
                        example.createCriteria().andEqualTo("orderId", orderId);
                        List<Order> orders = orderMapper.selectByExample(example);
                        //不是正常的数据就返回
                        if (orders.size() != 1) {
                            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                        }
                        Integer status = orders.get(0).getStatus();
                        //未付款说明过期了仍未付款，此时把order表 和库存表中状态都更新
                        if (status == 0) {
                            Order order = new Order();
                            //设为交易关闭
                            order.setStatus(5);
                            int i = orderMapper.updateByExampleSelective(order, example);
                            if (i != 1) {
                                log.info("DelayOrder 更新order表 出现问题");
                                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                            }
                            // 查询对应的商品id和 购买的数量
                            Example itemExample = new Example(OrderItem.class);
                            itemExample.createCriteria().andEqualTo("orderId", orderId);
                            List<OrderItem> orderItems = orderItemMapper.selectByExample(itemExample);
                            if (orderItems.size()==0){
                                log.info("OrderItem 表 出现问题");
                                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                            }
                            //每一个记录对应的商品id 和库存 然后去库存表中 进行加减
                            for (OrderItem orderItem : orderItems) {
                                Long itemId = orderItem.getItemId();
                                Integer num = orderItem.getNum();
                                Example stockExample = new Example(Stock.class);
                                stockExample.createCriteria().andEqualTo("itemId", itemId);
                                List<Stock> stocks = stockMapper.selectByExample(stockExample);
                                Integer lockCount = stocks.get(0).getLockCount();
                                Long stockCount = stocks.get(0).getStockCount();
                                Stock stock = new Stock();
                                stock.setStockCount(stockCount + num);
                                stock.setLockCount(lockCount - num);
                                int row1 = stockMapper.updateByExampleSelective(stock, stockExample);
                                if (row1 != 1) {
                                    log.info("更新库存表出现问题");
                                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                                }
                            }
                        }
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                    }
                    log.info("超时未支付");
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            //启动消费者
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}
